package com.lyon.custom.server.common.notify;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;

/**
 * @author LeeYan9
 * @since 2021-11-29
 */
@Slf4j
public class DefaultPublisher extends Thread implements Publisher {

    /**
     * 消费者
     */
    private List<Subscriber> subscribers;
    /**
     * 阻塞消息队列
     */
    private BlockingQueue<Event> queue;
    /**
     * 消费队列长度
     */
    private int bufferSize;
    private Executor executor;
    private Class<? extends Event> subscriberType;
    private volatile boolean shutdown;
    private volatile boolean initialized;

    @Override
    public void init(int bufferSize, Class<? extends Event> subscriberType) {
        this.subscriberType = subscriberType;
        this.subscribers = new CopyOnWriteArrayList<>();
        if (bufferSize < 1) {
            bufferSize = 1024;
        }
        this.bufferSize = bufferSize;
        queue = new ArrayBlockingQueue(bufferSize);
        setDaemon(true);
        setName("lyon-config-notify-");
        start();
    }

    @Override
    public void setExecutor(Executor executor) {
        this.executor = executor;
    }

    @Override
    public boolean shutdown() {
        return shutdown;
    }

    @Override
    public void addSubscriber(Subscriber consumer) {
        subscribers.add(consumer);
    }

    @Override
    public void publish(Event event) {
        Objects.nonNull(event);
        boolean success = queue.offer(event);
        if (!success) {
            log.warn("消息队列拥挤.. 准备同步消费队列");
            receiveEvent(event);
        }
    }

    private void receiveEvent(Event event) {
        for (Subscriber subscriber : subscribers) {
            try {

                if (executor != null) {
                    executor.execute(() -> subscriber.receiveEvent(event));
                } else {
                    ((Runnable) () -> subscriber.receiveEvent(event)).run();
                }
            } catch (Throwable e) {
                log.error(String.format("消息队列消费失败..[%s]", event.getClass()), e);
            }
        }
    }

    @Override
    public synchronized void start() {
        if (!initialized) {
            super.start();
            initialized = true;
        }
    }

    @Override
    public void run() {
        // 消费队列 ， 执行 subscribers
        for (; ; ) {
            if (shutdown() || CollUtil.isEmpty(subscribers)) {
                break;
            }
            ThreadUtil.sleep(1000);
            for (; ; ) {
                try {
                    Event event = queue.take();
                    receiveEvent(event);
                } catch (Throwable e) {
                    log.error("消费对列错误", e);
                }
            }
        }
    }
}
